/*
 * Created on 2006-8-4 上午01:17:30
 */
package cn.com.sbell.vss.csg.heartbeat;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.URL;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.httpclient.HostConfiguration;
import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.HttpHost;
import org.apache.commons.httpclient.HttpStatus;
import org.apache.commons.httpclient.NameValuePair;
import org.apache.commons.httpclient.methods.PostMethod;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.common.IoHandlerAdapter;
import org.apache.mina.common.IoSession;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.DocumentHelper;
import org.dom4j.Node;
import cn.com.sbell.vss.VssConstants;
import cn.com.sbell.vss.csg.vap.bean.AlarmMessage;
import cn.com.sbell.vss.csg.vap.util.VsInfoManager;
import cn.com.sbell.vss.csg.vsclient.util.CSGConfiguration;

public class HeartBeatHandler extends IoHandlerAdapter {

	private static Log logger = LogFactory.getLog(HeartBeatHandler.class);
	private static String CREATE_KEY = "create";
	private static int MAX_CACHE = 20000;

	public volatile AtomicLong holdNum = new AtomicLong(0L);
	public volatile long createNum = 0;
	public volatile long closeNum = 0;

	private Map<String, Long> registerCache;

	private HttpClient httpClient;
	private String cmsUrl;
	private HostConfiguration hostConfiguration;

	private long maxMemory = 0;
	private double freeMemoryLeastPercent = 0.1;

	public void init() {
		if (logger.isInfoEnabled()) {
			logger.info("Init HeartBeatHandler ...");
		}
		try {
			final URL url = new URL(cmsUrl);
			final String ip = url.getHost();
			final int port = url.getPort();
			hostConfiguration = new HostConfiguration();
			hostConfiguration.setHost(new HttpHost(ip, port));
		} catch (Exception ex) {
			if (logger.isWarnEnabled()) {
				logger.warn("Init HeartBeatHandler error:" + cmsUrl, ex);
			}
		}
		maxMemory = Runtime.getRuntime().maxMemory();
		logger.info("maxMemory=" + maxMemory / (1024 * 1024) + "m");
		if (CSGConfiguration.MINA_FREEMEM_PERCENT >= 0 && CSGConfiguration.MINA_FREEMEM_PERCENT <= 100) {
			freeMemoryLeastPercent = CSGConfiguration.MINA_FREEMEM_PERCENT;
			freeMemoryLeastPercent /= 100;
		}
		logger.info("freeMemoryLeastPercent=" + freeMemoryLeastPercent);
		registerCache = new HashMap<String, Long>();
	}

	public void exceptionCaught(IoSession session, Throwable throwable) throws Exception {
		if(logger.isDebugEnabled()) {
			logger.debug("exceptionCaught" + session,throwable);
		}
		session.close();
	}

	@Override
	public void sessionCreated(IoSession session) throws Exception {
		createNum++;
		long useMemory = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();
		logger.debug("useMemeory=" + useMemory / (1024 * 1024) + "m");
		if (((double) useMemory / maxMemory) >= (1 - freeMemoryLeastPercent)) {
			logger.warn("CSG nearly out of memory, request reject! Memory use/max=" + useMemory / (1024 * 1024) + "/" + maxMemory
					/ (1024 * 1024));
			session.close();
		} else {
			holdNum.incrementAndGet();
			session.setAttribute(CREATE_KEY, System.currentTimeMillis());
		}
		if (logger.isDebugEnabled()) {
			logger.debug("sessionCreated:createNum=" + createNum + ",holdNum+" + holdNum.longValue() + "," +  session);
		}
	}

	@Override
	public void sessionClosed(IoSession session) throws Exception {
		closeNum++;
		if (session.getAttribute(CREATE_KEY) != null) {
			holdNum.decrementAndGet();
		}
		if (logger.isDebugEnabled()) {
			logger.debug("sessionClosed:closeNum=" + closeNum + ",holdNum+" + holdNum.longValue() + "," +  session);
		}
	}

	public void messageReceived(IoSession session, Object message) throws Exception {
		//filter the session that hold too long, close directly
		long createTime = (Long) session.getAttribute(CREATE_KEY);
		if ((System.currentTimeMillis() - createTime) >= CSGConfiguration.MINA_THREADPOOL_HOLDTIME * 1000) {
			session.close();
			logger.debug("session hold too long, reject!");
			return;
		}

		if (logger.isDebugEnabled()) {
			logger.debug("messageReceived (" + session.getRemoteAddress() + ")");
		}

		if (message instanceof HeartBeatMessage) {
			SocketAddress addr = session.getRemoteAddress();

			HeartBeatMessage msg = (HeartBeatMessage) message;
			if (null == msg.getServerIp()) {
				// heartbeat message does not contain IP info, so we get it from connection
				if (addr instanceof InetSocketAddress) {
					msg.setServerIp(((InetSocketAddress) addr).getAddress().getHostAddress());
				}
			}
			if (0 == msg.getPort()) {
				// heartbeat message does not contain Port info, so we get it from connection
				if (addr instanceof InetSocketAddress) {
					msg.setPort(((InetSocketAddress) addr).getPort());
				}
			}
			// NAT
			msg.setOriginalIp(msg.getServerIp());
			msg.setOriginalPort(msg.getPort());
			if (addr instanceof InetSocketAddress) {
				msg.setServerIp(((InetSocketAddress) addr).getAddress().getHostAddress());
				msg.setPort(((InetSocketAddress) addr).getPort());
			}
			switch (msg.getMessageType()) {
			case HeartBeatMessage.REGISTER:
				if (logger.isInfoEnabled()) {
					logger.info("The register message from VS (NAT Version):\n" + msg.toString());
				}

				ResponseInfoToVS responseInfoToVS = new ResponseInfoToVS();
				responseInfoToVS.setRequestMessageType(HeartBeatMessage.REGISTER);

				boolean isReportOK = reportRegisterMessage(responseInfoToVS, msg);
				if (!isReportOK) {
					if (logger.isDebugEnabled()) {
						logger.debug("reportRegisterMessage() faild!(" + session.getRemoteAddress() + ")");
					}
				}
				if (msg.isNeedAcknowledge()) {
					// dynamic ip
					if (VssConstants.CODE_SUCCESS.equals(responseInfoToVS.getReturnCodeByEMS())) {
						if (null != msg.getServerId() && null != msg.getServerIp() && 0 != msg.getPort()) {
							if (logger.isDebugEnabled()) {
								logger.debug("HeartBeatHandler:addVsInfo:" + msg.toString());
							}
							// VsInfoManager.getVsInfoManager().addVsInfo(new VsInfo(msg.getServerId(),msg.getServerIp(),new
							// Integer(msg.getPort()),msg.getServerType()));
							// when dvs passed through irresponsible proxy, not NAT device, any information cached should be deleted.
							// obtain new mapping from CMS at next access.
							VsInfoManager.getVsInfoManager().deleteVsInfo(msg.getServerId());
						} else {
							if (logger.isDebugEnabled()) {
								logger.debug("HeartBeatHandler:VsInfoManager.getVsInfoManager().addVsInfo failed:" + msg.toString());
							}
						}
					}

					// join future, otherwise may fail to register
					session.write(responseInfoToVS).join(CSGConfiguration.PU_SO_TIMEOUT);
				}
				break;
			case HeartBeatMessage.HEARTBEAT:
				if (logger.isInfoEnabled()) {
					logger.info("The heartbeat message from VS (NAT Version):\n" + msg.toString());
				}
				processHeartBeat(msg);
				break;
			}
		}
		// dahua will not close session
		session.close();
	}

	private boolean inRegisterCache(HeartBeatMessage msg) {
		boolean result = false;
		synchronized(registerCache) {
			Long lastTime = registerCache.get(msg.getServerId());
			if(lastTime != null) {
				if( (System.currentTimeMillis() - lastTime) <= CSGConfiguration.VS_REGISTER_INTERVAL_MIN * 1000) {
					result = true;
				} else {
					result = false;
					registerCache.remove(msg.getServerId());
				}
			}
		}
		return result;
	}

	private void putRegisterCache(HeartBeatMessage msg) {
		synchronized(registerCache) {
			registerCache.put(msg.getServerId(), System.currentTimeMillis());
			if(registerCache.size() >= MAX_CACHE) {
				Map.Entry<String, Long> entry = null;
				Iterator<Map.Entry<String, Long>> iterator = registerCache.entrySet().iterator();
				while(iterator.hasNext()){
					entry = (Map.Entry<String, Long>)iterator.next();
					if( (System.currentTimeMillis() - entry.getValue()) >= CSGConfiguration.VS_REGISTER_INTERVAL_MIN * 1000) {
						iterator.remove();
					}
				}
			}
		}
	}

	private boolean reportRegisterMessage(ResponseInfoToVS responseInfoToVS, HeartBeatMessage msg) {
		if(inRegisterCache(msg)) {
			responseInfoToVS.setReturnCodeByEMS(ResponseInfoToVS.VS_REGISTER_DUP);
			return false;
		} else {
			putRegisterCache(msg);
		}
		boolean returnValue = true;
		int statusCode = -1;
		NameValuePair[] pairs = new NameValuePair[] { new NameValuePair("deviceId", msg.getServerId()),
				new NameValuePair("ip", msg.getServerIp()), new NameValuePair("port", String.valueOf(msg.getPort())),
				new NameValuePair("originalIp", msg.getOriginalIp()),
				new NameValuePair("originalPort", String.valueOf(msg.getOriginalPort())),
				new NameValuePair("accessPassword", msg.getAccessPassword()) };

		PostMethod method = new PostMethod(cmsUrl + "/RegisterVS");

		// add vondart
		if ((null != msg.getServerType()) && (AlarmMessage.SERVER_AH.equalsIgnoreCase(msg.getServerType()))) {
			method = new PostMethod(cmsUrl + "/RegisterAH");
			if (logger.isWarnEnabled()) {
				logger.warn("Send AH register message to " + cmsUrl + "/RegisterAH");
			}
		} else {
			if (logger.isWarnEnabled()) {
				logger.warn("Send VS register message to " + cmsUrl + "/RegisterVS");
			}
		}
		method.addParameters(pairs);

		try {
			// Execute the method.
			statusCode = httpClient.executeMethod(hostConfiguration, method);
			if (HttpStatus.SC_OK == statusCode) {
				String body = method.getResponseBodyAsString();
				Document doc = DocumentHelper.parseText(body);
				Node ntpIpNode = doc.selectSingleNode("//ntpIp");
				Node ntpPortNode = doc.selectSingleNode("//ntpPort");
				Node currentTimeNode = doc.selectSingleNode("//currentTime");
				Node timezoneNode = doc.selectSingleNode("//timezone");
				// ms username, password
				Node msPassWord = doc.selectSingleNode("//password");
				Node msUserName = doc.selectSingleNode("//username");
				Node returnCodeByEMSNode = doc.selectSingleNode("/response/result/@code");
				if (null != ntpIpNode) {
					responseInfoToVS.setNtpIp(ntpIpNode.getText());
				}
				if (null != ntpPortNode) {
					responseInfoToVS.setNtpPort(ntpPortNode.getText());
				}
				if (null != currentTimeNode) {
					responseInfoToVS.setCurrentTime(currentTimeNode.getText());
				}
				if (null != timezoneNode) {
					responseInfoToVS.setTimezone(timezoneNode.getText());
				}
				if (null != returnCodeByEMSNode) {
					int returnCodeByCSG = Integer.valueOf(returnCodeByEMSNode.getText());
					switch (returnCodeByCSG) {
					case ResponseInfoToVS.FROM_EMS_IP_PARAM_ERROR:
						responseInfoToVS.setReturnCodeByEMS(ResponseInfoToVS.FROM_CSG_IP_PARAM_ERROR);
						break;
					case ResponseInfoToVS.FROM_EMS_ACCESS_CODE_ERROR:
						responseInfoToVS.setReturnCodeByEMS(ResponseInfoToVS.FROM_CSG_ACCESS_CODE_ERROR);
						break;

					case ResponseInfoToVS.FROM_EMS_VS_ID_NONEXIST:
						responseInfoToVS.setReturnCodeByEMS(ResponseInfoToVS.FROM_CSG_VS_ID_NONEXIST);
						break;
					case ResponseInfoToVS.FROM_EMS_OK:
						responseInfoToVS.setReturnCodeByEMS(ResponseInfoToVS.FROM_CSG_OK);
						break;
					}

					if (logger.isInfoEnabled()) {
						logger.info("ReturnCodeByEMS:" + returnCodeByEMSNode.getText());
					}
				}
				if (null != msUserName) {
					responseInfoToVS.setMsUserName(msUserName.getText());
					if (logger.isInfoEnabled()) {
						logger.info("MS UserName:" + msUserName.getText());
					}
				}
				if (null != msPassWord) {
					responseInfoToVS.setMsPassWord(msPassWord.getText());
					if (logger.isInfoEnabled()) {
						logger.info("MS PassWord:" + msPassWord.getText());
					}
				}
				responseInfoToVS.setHttpResponseStatusOKByEMS(true);
			}
		} catch (IOException e) {
			if (logger.isWarnEnabled()) {
				logger.warn("throws IOException in reportRegisterMessage()." + pairs.toString());
			}
		} catch (DocumentException e) {
			if (logger.isWarnEnabled()) {
				logger.warn("throws DocumentException in reportRegisterMessage()." + pairs.toString());
			}
		} finally {
			if (HttpStatus.SC_OK != statusCode) {
				if (logger.isWarnEnabled()) {
					logger.warn("PostMethod failed: " + method.getStatusLine());
				}
				
				responseInfoToVS.setReturnCodeByEMS(ResponseInfoToVS.EMS_ERROR);
				returnValue = false;
			}
			method.releaseConnection();
		}

		return returnValue;
	}

	private void processHeartBeat(HeartBeatMessage msg) {
		if (msg.getServerId() != null && !msg.getServerId().equals("")) {
			HeartBeatMonitor.getMonitorInstance().update(msg);
		}
	}

	/**
	 * @return the emsUrl
	 */
	public String getCmsUrl() {
		return cmsUrl;
	}

	/**
	 * @param emsUrl
	 *            the emsUrl to set
	 */
	public void setCmsUrl(String cmsUrl) {
		this.cmsUrl = cmsUrl;
	}

	/**
	 * @return the httpClient
	 */
	public HttpClient getHttpClient() {
		return httpClient;
	}

	/**
	 * @param httpClient
	 *            the httpClient to set
	 */
	public void setHttpClient(HttpClient httpClient) {
		this.httpClient = httpClient;
	}

}
